package org.example.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.metrics.KafkaMetric;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.Scanner;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public class KafkaFastTest {
    public static final String brokerList = "192.168.204.117:9092";

    public static final String topic = "topic-demo";

    public static final AtomicBoolean isRunning = new AtomicBoolean(true);

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, FastJsonSerializer.class);

        KafkaProducer producer = new KafkaProducer(properties);

        // 发送消息
        Scanner scanner = new Scanner(System.in);
        while (isRunning.get()) {
            String text = scanner.next();
            if (StringUtils.isNotBlank(text)) {
                String[] content = text.split(",");
                User user = new User();
                user.setFirstName(content[0]);
                user.setLastName(content[1]);
                user.setAge(Integer.parseInt(content[2]));
                user.setId(Long.parseLong(content[3]));

                ProducerRecord<String, User> record = new ProducerRecord<>(topic, user);
                producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        if (e == null) {
                            log.info("partition'num = " + recordMetadata.partition() + ", offset = " + recordMetadata.offset());
                            // 打印该topic有哪些分区
                            List list = producer.partitionsFor(topic);
                            log.info("分区：" + Arrays.toString(list.toArray()));
                        } else {
                            log.error(e.getMessage(), e);
                        }
                    }
                });
            }
        }
        producer.close();
    }
}
